package graphframes

import org.apache.spark.sql.types._
import org.apache.spark.sql.{Column, SparkSession}
import org.graphframes.GraphFrame

import scala.reflect.io.Directory

object bfs extends App {
  val spark = SparkSession.builder()
    .appName("bfs-scala")
    .master("local[2]")
    .getOrCreate()
  //获取节点
  val gNodes = spark.read.format("csv")
    .schema(StructType(Array(
      StructField("id", IntegerType, nullable = true),
      StructField("name", StringType, nullable = true),
      StructField("latitude", FloatType, nullable = true),
      StructField("longitude", FloatType, nullable = true),
      StructField("population", IntegerType, nullable = true))))
    .option("header", "true")
    .load(resourcesPath + "/data/transport-nodes.csv")
  //获取关系 (这里简化了指定schema)
  val gRels = spark.read.format("csv")
    .option("header", "true")
    .load(resourcesPath + "/data/transport-relationships.csv")

  gNodes.show()
  //将关系翻转
  //翻转原因如下：道路的关系是双向的 但是spark和neo4j只能处理单相关系。
  //因此为了描述双向关系在src和dst节点之间添加了两个关系
  val gReversedRels = gRels.withColumn("newSrc", gRels.col("dst"))
    .withColumn("newDst", gRels.col("src"))
    .drop("dst", "src")
    .withColumnRenamed("newSrc", "src")
    .withColumnRenamed("newDst", "dst")
    .select("src", "dst", "relationship", "cost")
  //合成完整的关系
  val gRelationships = gRels.union(gReversedRels)
  //生成图（这里生成的是GraphFrame 其数据为DataFrame 而GraphX基于RDD）
  val graph = GraphFrame.apply(gNodes, gRelationships)
  gRelationships.show()
  //有一定条件的广度搜索
  //该方法返回的数据是列存储的
  val result = graph.bfs.fromExpr("id=3")
    .toExpr("population > 100000 and population < 300000 and id <> 3")
    .run()
  graph.persist() //这里将图形缓存 提升效率


  //基于GraphFrame 的优点是便于实现基于检索条件的过滤 比如检索某些条件的Node
  graph.vertices.filter("population > 100000 and population < 300000")
    .sort("population")
    .show()
  //将检索对node进行过滤
  val columns = result.columns.filter(!_.startsWith("e")).map(new Column(_))
  //缓存结果提高效率
  result.persist()
  result.show()
  //计算resources目录的绝对路径
  var resourcesPath = "file:///" + Directory.Current.get.toString()
  result.select(columns: _*).show()

  //删除缓存
  result.unpersist()
  graph.unpersist()

  //Note!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
  //该搜索算法是达到最先满足的顶点就直接结束了
  //该例中满足 "population > 100000 and population < 300000 and id <> 'Den Haag'"条件的 Node为
  //Colchester和Ipswich 但由于Ipswich更近 所有搜索得到的结果是 'Den Haag'和Ipswich之间的最短路径
}
